package com.deep.flink;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Calendar;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author sw
 * @create 2023-01-18 13:15
 */
public class ClickSource implements SourceFunction<Event> {

    private boolean running = true;

    private AtomicInteger counter =  new AtomicInteger(1);

    @Override
    public void run(SourceContext<Event> ctx) throws Exception {
        Random random = new Random();
        Integer[] ages = {1, 2, 3, 4};
        String[] users = {"mary", "Alice", "Bob", "Cary"};
        String[] urls = {"./home", "./cart", "./prod?id=100", "./prod?id=10"};

        while (running) {
            if (counter.get() > 20) {
                running = false;
                break;
            }
            String username = users[random.nextInt(users.length)];
            String url = urls[random.nextInt(urls.length)];
            Long timestamp = Calendar.getInstance().getTimeInMillis();
            Integer age = ages[random.nextInt(ages.length)];

            ctx.collect(new Event(username, url, timestamp, age));

            counter.incrementAndGet();

            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        running = false;
    }
}
